package v08

import (
	"context"
	"errors"
	"time"

	"git.edaijia-inc.cn/edaijia/ego/ekafka/consumer"
	"git.edaijia-inc.cn/edaijia/ego/elogger"
	eloggerkafka "git.edaijia-inc.cn/edaijia/ego/elogger/kafka"

	"github.com/Shopify/sarama"
	consumergroup "github.com/meitu/go-consumergroup"
)

type EKafkaV08ConsumerGroup struct {
	Client *consumergroup.ConsumerGroup
	Conf   *consumer.EKafkaConsumerConfig
}

func NewEKafkaV08ConsumerGroup(conf *consumer.EKafkaConsumerConfig) (*EKafkaV08ConsumerGroup, error) {
	if conf.Group == "" {
		return nil, errors.New("Group Name Error")
	}
	if len(conf.Topics) == 0 {
		return nil, errors.New("Topics is empty")
	}
	if conf.Workers <= 0 {
		return nil, errors.New("Number of Workers is zero")
	}

	//默认0.8.2.1
	if conf.Version == "" {
		conf.Version = "0.8.2.1"
	}

	if conf.CommitInterval == 0 {
		conf.CommitInterval = 2
	}

	var err error
	eloggerkafka.InitELoggerKafka(conf.Verbose)

	cgConf := consumergroup.NewConfig()
	cgConf.ZkList = conf.Zookeepers
	cgConf.ZkSessionTimeout = 6 * time.Second
	cgConf.TopicList = conf.Topics
	cgConf.GroupID = conf.Group
	cgConf.OffsetAutoCommitEnable = !conf.MannulCommit
	cgConf.OffsetAutoCommitInterval = time.Duration(conf.CommitInterval) * time.Second
	cgConf.SaramaConfig.Version, err = sarama.ParseKafkaVersion(conf.Version)
	if err != nil {
		elogger.Error("Kafka version error:", err)
		return nil, err
	}

	cg, err := consumergroup.NewConsumerGroup(cgConf)
	if err != nil {
		elogger.Error("Failed to create consumer group, err ", err)
		return nil, err
	}

	cg.SetLogger(elogger.GetELogger())

	err = cg.Start()
	if err != nil {
		elogger.Error("Failed to join group, err ", err)
		return nil, err
	}

	// Retrieve the error and log
	for _, tp := range conf.Topics {
		go func(topic string) {
			if topicErrChan, ok := cg.GetErrors(topic); ok {
				for err := range topicErrChan {
					if err != nil {
						elogger.Error("Toipic ", topic, " got err, ", err)
					}
				}
			}
		}(tp)
	}

	return &EKafkaV08ConsumerGroup{cg, conf}, nil
}

func (e *EKafkaV08ConsumerGroup) GetMessage(ctx context.Context, handle consumer.HandleFunc) {
	for _, tp := range e.Conf.Topics {
		for i := 0; i < e.Conf.Workers; i++ {
			go func(topic string) {
				elogger.Info("Group ", e.Conf.Group, " start worker to get message from topic ", topic)
				if msgChan, ok := e.Client.GetMessages(topic); ok {
					for message := range msgChan {
						handle(message.Topic, string(message.Value), message.Partition, message.Offset)
						if e.Conf.MannulCommit {
							e.Client.CommitOffset(message.Topic, message.Partition, message.Offset)
						}
					}
				} else {
					elogger.Info("Topic ", topic, " was not found in consumergroup")
				}
			}(tp)
		}
	}

	//TODO 调用Stop,提交commit
}
